package com.ruoyi.kafka.kafka;

import com.ruoyi.kafka.domain.MessageEnvelope;
import com.ruoyi.kafka.domain.OrderCreatedEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.UUID;

/**
 * @ClassName OrderProducer
 * @Description Kafka 生产者服务（发送订单创建事件）
 * @Author li yang
 * @Date 2025/9/18 16:46
 * @Version 1.0
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderProducer {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void send(OrderCreatedEvent event) {
        MessageEnvelope<OrderCreatedEvent> envelope = MessageEnvelope.of(
                event.getOrderId(),
                "ORDER_CREATED",
                "v1",
                "ly-service",
                UUID.randomUUID().toString(),
                event
        );

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future =
                kafkaTemplate.send("orders", event.getOrderId(), envelope);

        // 发送结果回调
        future.addCallback(
                result -> {
                    RecordMetadata md = null;
                    if (result != null) {
                        md = result.getRecordMetadata();
                    }
                    if (md != null) {
                        log.info("发送成功: topic={} partition={} offset={}",
                                md.topic(), md.partition(), md.offset());
                    }
                },
                ex -> log.error("发送失败: {}", ex.getMessage())
        );
    }
}
